package cn.ultrarisk.scadmin.manager.biz.solr.service;

import cn.ultrarisk.mybatis.extension.query.Query;
import cn.ultrarisk.mybatis.extension.query.Updater;
import cn.ultrarisk.scadmin.manager.biz.common.context.OperationType;
import cn.ultrarisk.scadmin.manager.biz.common.service.OperationLogService;
import cn.ultrarisk.scadmin.manager.biz.solr.dao.SolrCollectionDAO;
import cn.ultrarisk.scadmin.manager.biz.solr.dao.SolrReplicaDAO;
import cn.ultrarisk.scadmin.manager.biz.solr.dao.SolrShardDAO;
import cn.ultrarisk.scadmin.manager.biz.solr.dto.CollectionSchema;
import cn.ultrarisk.scadmin.manager.biz.solr.entity.SolrCollection;
import cn.ultrarisk.scadmin.manager.biz.solr.entity.SolrReplica;
import cn.ultrarisk.scadmin.manager.biz.solr.entity.SolrShard;
import cn.ultrarisk.scadmin.manager.biz.solr.factory.CloudSolrServerFactory;
import cn.ultrarisk.scadmin.manager.biz.zookeeper.dao.ZkNodeDAO;
import cn.ultrarisk.scadmin.manager.biz.zookeeper.entity.ZkNode;
import cn.ultrarisk.scadmin.manager.biz.zookeeper.service.ZkNodeService;
import cn.ultrarisk.scadmin.manager.exception.BizCheckedException;
import cn.ultrarisk.scadmin.manager.util.CollectionUtil;
import cn.ultrarisk.scadmin.manager.util.JsonUtil;
import org.apache.log4j.Logger;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.impl.CloudSolrServer;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * SolrCollectionService
 *
 * @author sevendlong(lin23871@163.com)
 * @date 2014/5/13
 * @since V1.0
 */
@Service
@Transactional(propagation = Propagation.NOT_SUPPORTED)
public class SolrCollectionService {
    private static Logger log = Logger.getLogger(SolrCollectionService.class);

    @Autowired
    private SolrCollectionDAO solrCollectionDAO;
    @Autowired
    private SolrShardDAO solrShardDAO;
    @Autowired
    private ZkNodeDAO zkNodeDAO;
    @Autowired
    private ZkNodeService zkNodeService;
    @Autowired
    private SolrReplicaDAO solrReplicaDAO;
    @Autowired
    private SolrReplicaService solrReplicaService;
    @Autowired
    private OperationLogService operationLogService;
    @Autowired
    private CloudSolrServerFactory solrServerFactory;

    private void createShardReplicas(SolrCollection collection, Map<String, List<SolrReplica>> addShardReplicas) {
        if (addShardReplicas != null && addShardReplicas.size() > 0) {
            for (String shardName : addShardReplicas.keySet()) {
                List<SolrReplica> replicas = addShardReplicas.get(shardName);
                if (replicas != null && replicas.size() > 0) {
                    for (int i = 0; i < replicas.size(); i++) {
                        String coreName = replicas.get(i).getCoreName();
                        Integer nodeId = replicas.get(i).getNodeId();
                        solrReplicaService.create(coreName, collection.getId(), shardName, nodeId, false);
                    }
                }
            }
        }
    }

    public void create(String name, String config, Boolean allowCommit, Boolean allowRollback, Integer maxStart, Integer maxLimit,
                       String remark, Map<String, List<SolrReplica>> shardReplicas) {
        ZkNode configNode = zkNodeDAO.getByPath(config, 0);
        ZkNode schemaNode = zkNodeDAO.getByPath(config + "/schema.xml", 0);
        if (configNode == null || schemaNode == null) {
            throw new BizCheckedException("config is empty");
        }

        SolrCollection collection = solrCollectionDAO.create(name, config, allowCommit, allowRollback, maxStart, maxLimit, remark);

        CollectionSchema schema = CollectionSchema.fromXml(schemaNode.getData());
        updateUniqueKey(collection.getId(), schema.getUniqueKey());

        zkNodeService.create("/collections/" + name, "{\"configName\":\"" + configNode.getRelativePath() + "\"}", 0);
        createShardReplicas(collection, shardReplicas);

        operationLogService.createSolrCollectionLog(OperationType.SOLR_COLLECTION_CREATE, collection.getId());
    }

    public void update(Integer id, Boolean allowCommit, Boolean allowRollback, Integer maxStart, Integer maxLimit,
                       String remark, Map<String, List<SolrReplica>> shardReplicas) {
        solrCollectionDAO.update(id, allowCommit, allowRollback, maxStart, maxLimit, remark);
        SolrCollection collection = solrCollectionDAO.get(id);

        Map<String, SolrReplica> addReplicaMap = new HashMap<String, SolrReplica>();
        for (String shardName : shardReplicas.keySet()) {
            if (shardReplicas.get(shardName) != null) {
                for (SolrReplica replica : shardReplicas.get(shardName)) {
                    String replicaKey = CollectionUtil.join(new String[]{shardName, replica.getNodeId().toString(), replica.getCoreName()}, "-");
                    addReplicaMap.put(replicaKey, replica);
                }
            }
        }

        List<SolrShard> existShards = solrShardDAO.findAvailable(id);
        List<SolrReplica> existReplicas = solrReplicaDAO.findAvailable(null, id, false);
        Map<Integer, String> existShardMap = new HashMap<Integer, String>();
        for (SolrShard existShard : existShards) {
            existShardMap.put(existShard.getId(), existShard.getName());
        }
        Map<String, SolrReplica> existReplicaMap = new HashMap<String, SolrReplica>();
        for (SolrReplica existReplica : existReplicas) {
            String replicaKey = CollectionUtil.join(new String[]{existShardMap.get(existReplica.getShardId()), existReplica.getNodeId().toString(), existReplica.getCoreName()}, "-");
            existReplicaMap.put(replicaKey, existReplica);
        }

        List<Integer> removeShardIds = new ArrayList<Integer>();
        List<Integer> removeReplicaIds = new ArrayList<Integer>();
        for (Integer existShardId : existShardMap.keySet()) {
            String existShardName = existShardMap.get(existShardId);
            if (!shardReplicas.containsKey(existShardName)) {
                removeShardIds.add(existShardId);
            }
        }
        for (String existReplicaKey : existReplicaMap.keySet()) {
            if (!addReplicaMap.containsKey(existReplicaKey)) {
                removeReplicaIds.add(existReplicaMap.get(existReplicaKey).getId());
            }
        }

        if (removeReplicaIds.size() > 0) {
            for (Integer removeReplicaId : removeReplicaIds) {
                try {
                    solrReplicaService.delete(removeReplicaId, false);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if (removeShardIds.size() > 0) {
                solrShardDAO.delete(removeShardIds);
            }
        }

        Map<String, List<SolrReplica>> addShardReplicas = new HashMap<String, List<SolrReplica>>();
        for (String addReplicaKey : addReplicaMap.keySet()) {
            String addShardName = addReplicaKey.split("-")[0];
            if (!existReplicaMap.containsKey(addReplicaKey)) {
                List<SolrReplica> addReplicas = addShardReplicas.get(addShardName);
                if (addReplicas == null) {
                    addReplicas = new ArrayList<SolrReplica>();
                }
                addReplicas.add(addReplicaMap.get(addReplicaKey));
                addShardReplicas.put(addShardName, addReplicas);
            }
        }
        createShardReplicas(collection, addShardReplicas);

        operationLogService.createSolrCollectionLog(OperationType.SOLR_COLLECTION_UPDATE, collection.getId());
    }

    public void optimize(Integer id, Integer maxSegments) throws IOException, SolrServerException {
        SolrCollection collection = solrCollectionDAO.get(id, SolrCollection.NAME);
        CloudSolrServer server = solrServerFactory.getServer();
        server.setDefaultCollection(collection.getName());
        server.optimize(false, false, maxSegments);
        operationLogService.createSolrCollectionLog(OperationType.SOLR_COLLECTION_OPTIMIZE, id, maxSegments);
    }

    public void reload(Integer id) throws IOException, SolrServerException {
        SolrCollection collection = solrCollectionDAO.get(id, SolrCollection.NAME, SolrCollection.CONFIG);
        CloudSolrServer server = solrServerFactory.getServer();
        CollectionAdminRequest.reloadCollection(collection.getName(), server);
        operationLogService.createSolrCollectionLog(OperationType.SOLR_COLLECTION_RELOAD, id);

        CollectionSchema schema = solrCollectionDAO.getSchemaFromZk(id);
        updateUniqueKey(id, schema.getUniqueKey());
    }

    public void delete(Integer id) throws IOException, SolrServerException {
        SolrCollection collection = solrCollectionDAO.get(id, SolrCollection.NAME, SolrCollection.AVAILABLE);

        if (collection.getAvailable()) {
            CollectionAdminRequest.deleteCollection(collection.getName(), solrServerFactory.getServer());
        }

        operationLogService.createSolrCollectionLog(OperationType.SOLR_COLLECTION_DELETE, id);

        solrReplicaDAO.delete(Query.query(SolrReplica.COLLECTION_ID, id));
        solrShardDAO.delete(Query.query(SolrShard.COLLECTION_ID, id));
        solrCollectionDAO.delete(id);
    }

    public void updateRouter(Integer id, JsonUtil.MapWrapper collectionStateMap) {
        String router = collectionStateMap.getStringByPath("router.name");
        solrCollectionDAO.update(id, Updater.update(SolrCollection.ROUTER, router));
    }

    public void updateUniqueKey(Integer id, String uniqueKey) {
        SolrCollection collection = solrCollectionDAO.get(id, SolrCollection.NAME);
        solrCollectionDAO.update(id, Updater.update(SolrCollection.UNIQUE_KEY, uniqueKey));
    }

    public void updateAvailable(Integer id, Boolean available) {
        solrCollectionDAO.update(id, Updater.update(SolrCollection.AVAILABLE, available));
    }

    public SolrCollection saveByZKInfo(String name, String path, String data, String uniqueKey, boolean available, Long updateTime) {
        JsonUtil.MapWrapper dataMap = new JsonUtil.MapWrapper(JsonUtil.fromJsonAsMap(data));
        String config = "/configs/" + dataMap.getStringByPath("configName");
        SolrCollection collection = solrCollectionDAO.get(Query.query(SolrCollection.NAME, name));
        if (collection == null) {
            collection = new SolrCollection(name, config);
            collection.setAllowCommit(false);
            collection.setAllowRollback(false);
        }

        collection.setAvailable(available);
        collection.setUniqueKey(uniqueKey);
        collection.setRemark("");
        collection.setUpdateTime(updateTime);

        if (collection.getId() == null) {
            solrCollectionDAO.create(collection);
        } else {
            solrCollectionDAO.update(collection.getId(),
                    Updater.update(SolrCollection.CONFIG, config).addField(SolrCollection.AVAILABLE, available)
                            .addField(SolrCollection.UPDATE_TIME, updateTime).addField(SolrCollection.UNIQUE_KEY, uniqueKey));
        }
        return collection;
    }

    public void disable(List<Integer> ids) {
        solrCollectionDAO.update(Query.inQuery(SolrCollection.ID, ids), Updater.update(SolrCollection.AVAILABLE, false));
    }

    /**
     * 清除过期collection
     */
    public void clearExpiredData(long expiredTime) {
        Query query = Query.ltQuery(SolrCollection.UPDATE_TIME, expiredTime);
        solrCollectionDAO.delete(query);
    }
}
